package com.blyat.xsoft.kernel.component.mq.rocketmq.producer.transactor;

import com.alibaba.fastjson.JSON;
import com.blyat.xsoft.kernel.component.mq.rocketmq.Carrier;
import com.blyat.xsoft.kernel.component.mq.rocketmq.RocketMQContextLoader;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Auther: syh
 * @Date: 2020/8/7
 * @Description:
 */
public abstract class AbstractRocketMQTransactor<T> extends RocketMQContextLoader implements TransactionListener {
    private static final int DEFAULT_IO_THREADS = Math.min(Runtime.getRuntime().availableProcessors() + 1, 32);
    private static final int DEFAULT_KEEP_ALIVE_TIME = 60 * 1000;
    private static final int DEFAULT_BLOCKING_QUEUE_SIZE = 2000;

    private TransactionTemplate transactionTemplate;
    private TransactionMQProducer producer;
    private int corePoolSize;
    private int maximumPoolSize;
    private int keepAliveTime;
    private int blockingQueueSize;

    public AbstractRocketMQTransactor() {
        this(DEFAULT_IO_THREADS);
    }

    public AbstractRocketMQTransactor(int corePoolSize) {
        this(corePoolSize, DEFAULT_IO_THREADS);
    }

    public AbstractRocketMQTransactor(int corePoolSize, int maximumPoolSize) {
        this(corePoolSize, maximumPoolSize, DEFAULT_KEEP_ALIVE_TIME, DEFAULT_BLOCKING_QUEUE_SIZE);
    }

    public AbstractRocketMQTransactor(int corePoolSize, int maximumPoolSize, int keepAliveTime, int blockingQueueSize) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.keepAliveTime = keepAliveTime;
        this.blockingQueueSize = blockingQueueSize;
    }

    public void setProducer(DefaultMQProducer producer) {
        this.producer = (TransactionMQProducer) producer;
    }

    @Override
    protected void doAfterSingletonsInstantiated() {
        this.transactionTemplate = getApplicationContext().getBean(TransactionTemplate.class);
        setTransactionListener();
    }

    private void setTransactionListener() {
        if (producer == null) {
            RocketMQTemplate rocketMQTemplate = getApplicationContext().getBean(RocketMQTemplate.class);
            producer = (TransactionMQProducer) rocketMQTemplate.getProducer();
        }

        if (producer.getTransactionListener() != null) {
            return;
        }
        producer.setExecutorService(new ThreadPoolExecutor(corePoolSize, maximumPoolSize,
                keepAliveTime, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(blockingQueueSize)));

        producer.setTransactionListener(this);
    }

    /**
     * 正常事务发送与提交阶段
     * 1、生产者发送一个半消息给MQServer（半消息是指消费者暂时不能消费的消息）
     * 2、服务端响应消息写入结果，半消息发送成功
     * 3、开始执行本地事务
     * 4、根据本地事务的执行状态执行Commit或者Rollback操作
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
        Carrier<T> carrier = JSON.parseObject(new String(message.getBody()), Carrier.class);
        // 执行本地事务begin
        Exception exception = transactionTemplate.execute(new TransactionCallback<Exception>() {
            public Exception doInTransaction(TransactionStatus status) {
                try {
                    AbstractRocketMQTransactor.this.execute(carrier.getContent());
                } catch (Exception e) {//设置事务回滚
                    status.setRollbackOnly();
                    return new Exception(e);
                }
                return null;
            }
        });

        if (exception != null) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

        return LocalTransactionState.COMMIT_MESSAGE;
    }

    /**
     * 事务信息的补偿流程
     * 1、如果MQServer长时间没收到本地事务的执行状态会向生产者发起一个确认回查的操作请求
     * 2、生产者收到确认回查请求后，检查本地事务的执行状态
     * 3、根据检查后的结果执行Commit或者Rollback操作
     * 补偿阶段主要是用于解决生产者在发送Commit或者Rollback操作时发生超时或UNKNOW的情况
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt message) {
        try {
            check(null);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    protected abstract void execute(T content) throws Exception;

    protected abstract void check(T content) throws Exception;
}
